package com.parsechina.hpush.broker.spi.persistence.redis;


import com.parsechina.hpush.broker.config.MqttConstants;
import com.parsechina.hpush.broker.internal.ClientSession;
import com.parsechina.hpush.broker.internal.MessageGUID;
import com.parsechina.hpush.broker.internal.PersistentSession;
import com.parsechina.hpush.broker.internal.StoredMessage;
import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.spi.IMessagesStore;
import com.parsechina.hpush.broker.spi.ISessionsStore;
import com.parsechina.hpush.broker.spi.persistence.SessionsStoreUtils;
import com.parsechina.hpush.broker.spi.utils.Utils;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * ISessionsStore implementation backed by Redis.
 *
 * @author linfeng
 */
class RedisSessionsStore implements ISessionsStore {

    private static final Logger LOG = LoggerFactory.getLogger(RedisSessionsStore.class);

    //maps clientId->[MessageId -> guid]
    private ConcurrentMap<String, ConcurrentMap<Integer, MessageGUID>> m_inflightStore;
    //map clientId <-> set of currently in flight packet identifiers
    private Map<String, Set<Integer>> m_inFlightIds;
    private ConcurrentMap<String, PersistentSession> m_persistentSessions;
    //maps clientId->[guid*], insertion order cares, it's queue
    private ConcurrentMap<String, List<MessageGUID>> m_enqueuedStore;
    //maps clientId->[MessageId -> guid]
    private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;

    private final RedissonClient redissonClient;
    private final IMessagesStore messagesStore;

    RedisSessionsStore(RedissonClient redissonClient, IMessagesStore messagesStore) {
        this.redissonClient = redissonClient;
        this.messagesStore = messagesStore;
    }

    @Override
    public void initStore() {
        m_inflightStore = redissonClient.getMap(MqttConstants.LOCAL_CACHE_INFLIGHT_KEY);
        m_inFlightIds = redissonClient.getMap(MqttConstants.LOCAL_CACHE_INFLIGHT_PACKET_IDS_KEY);
        m_persistentSessions = redissonClient.getMap(MqttConstants.LOCAL_CACHE_SESSIONS_KEY);
        m_enqueuedStore = redissonClient.getMap(MqttConstants.LOCAL_CACHE_SESSION_QUEUE_KEY);
        m_secondPhaseStore = redissonClient.getMap(MqttConstants.LOCAL_CACHE_SECOND_PHASE_KEY);
    }

    @Override
    public void addNewSubscription(Subscription newSubscription) {
        LOG.debug("addNewSubscription invoked with subscription {}", newSubscription);
        final String clientId = newSubscription.getClientId();
        redissonClient.getMap("subscriptions_" + clientId).put(newSubscription.getTopicFilter(), newSubscription);

        if (LOG.isTraceEnabled()) {
            LOG.trace("subscriptions_{}: {}", clientId, redissonClient.getMap("subscriptions_" + clientId));
        }
    }

    @Override
    public void removeSubscription(String topicFilter, String clientId) {
        LOG.debug("removeSubscription topic filter: {} for clientId: {}", topicFilter, clientId);
        RMap<String, Object> objectRMap = redissonClient.getMap("subscriptions_" + clientId);
        if (!objectRMap.isExists()) {
            return;
        }
        objectRMap.remove(topicFilter);
    }

    @Override
    public void removeAllSubscriptions(String sessionId) {
        LOG.debug("removeAllSubscriptions");
        RMap<String, Object> objectRMap = redissonClient.getMap("subscriptions_" + sessionId);
        if (objectRMap.isExists()) {
            objectRMap.delete();
        }
    }

    @Override
    public List<ClientTopicRelation> listAllSubscriptions() {
        final List<ClientTopicRelation> allSubscriptions = new ArrayList<>();
        for (String clientID : m_persistentSessions.keySet()) {
            ConcurrentMap<String, Subscription> clientSubscriptions = redissonClient.getMap("subscriptions_" + clientID);
            for (String topicFilter : clientSubscriptions.keySet()) {
                allSubscriptions.add(new ClientTopicRelation(clientID, topicFilter));
            }
        }
        LOG.debug("retrieveAllSubscriptions returning subs {}", allSubscriptions);
        return allSubscriptions;
    }

    @Override
    public Subscription getSubscription(ClientTopicRelation relation) {
        ConcurrentMap<String, Subscription> clientSubscriptions = redissonClient.getMap("subscriptions_" + relation.clientId);
        LOG.debug("subscriptions_{}: {}", relation.clientId, clientSubscriptions);
        return clientSubscriptions.get(relation.topicFilter);
    }

    @Override
    public List<Subscription> getSubscriptions() {
        List<Subscription> subscriptions = new ArrayList<>();
        for (String clientId : m_persistentSessions.keySet()) {
            ConcurrentMap<String, Subscription> clientSubscriptions = redissonClient.getMap("subscriptions_" + clientId);
            subscriptions.addAll(clientSubscriptions.values());
        }
        return subscriptions;
    }

    @Override
    public boolean contains(String clientId) {
        RMap<String, Object> tmpMap = redissonClient.getMap("subscriptions_" + clientId);
        return tmpMap.isExists();
    }

    @Override
    public ClientSession createNewSession(String clientId, boolean cleanSession) {
        LOG.debug("createNewSession for client <{}> with clean flag <{}>", clientId, cleanSession);
        if (m_persistentSessions.containsKey(clientId)) {
            LOG.error("already exists a session for client <{}>, bad condition", clientId);
            throw new IllegalArgumentException("Can't create a session with the ID of an already existing" + clientId);
        }
        LOG.debug("clientId {} is a newcome, creating it's empty subscriptions set", clientId);
        m_persistentSessions.putIfAbsent(clientId, new PersistentSession(cleanSession));
        return new ClientSession(clientId, messagesStore, this, cleanSession);
    }

    @Override
    public ClientSession sessionForClient(String clientId) {
        if (!m_persistentSessions.containsKey(clientId)) {
            return null;
        }

        PersistentSession storedSession = m_persistentSessions.get(clientId);
        return new ClientSession(clientId, messagesStore, this, storedSession.cleanSession);
    }

    @Override
    public void removeSessionOfClient(String clientId) {
        m_inflightStore.remove(clientId);
        m_inFlightIds.remove(clientId);
        m_persistentSessions.remove(clientId);
        m_enqueuedStore.remove(clientId);
        m_secondPhaseStore.remove(clientId);
        RMap<String, Object> tmpMap = redissonClient.getMap("subscriptions_" + clientId);
        if (tmpMap.isExists()) {
            tmpMap.delete();
        }
    }

    @Override
    public void updateCleanStatus(String clientId, boolean cleanSession) {
        m_persistentSessions.put(clientId, new PersistentSession(cleanSession));
    }

    /**
     * Return the next valid packetIdentifier for the given client session.
     */
    @Override
    public int nextPacketId(String clientId) {
        Set<Integer> inFlightForClient = m_inFlightIds.get(clientId);
        if (inFlightForClient == null) {
            inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<Integer, Boolean>());
            int nextPacketId = 1;
            inFlightForClient.add(nextPacketId);
            m_inFlightIds.put(clientId, inFlightForClient);
            return nextPacketId;

        }

        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
        int nextPacketId = (maxId + 1) % 0xFFFF;
        inFlightForClient.add(nextPacketId);
        return nextPacketId;
    }

    @Override
    public MessageGUID inFlightAck(String clientId, int messageId) {
        LOG.info("acknowledging inflight clientId <{}> messageID {}", clientId, messageId);
        Map<Integer, MessageGUID> m = m_inflightStore.get(clientId);
        if (m == null || m.isEmpty()) {
            LOG.error("Can't find the inFlight record for client <{}>", clientId);
            return null;
        }
        MessageGUID guid = m.remove(messageId);

        //remove from the ids store
        Set<Integer> inFlightForClient = m_inFlightIds.get(clientId);
        if (inFlightForClient != null && !inFlightForClient.isEmpty()) {
            inFlightForClient.remove(messageId);
        }
        return guid;

    }

    @Override
    public void inFlight(String clientId, int messageId, MessageGUID guid) {
        ConcurrentMap<Integer, MessageGUID> m = m_inflightStore.get(clientId);
        if (m == null) {
            m = new ConcurrentHashMap<>();
        }
        m.put(messageId, guid);
        LOG.info("storing inflight clientId <{}> messageID {} guid <{}>", clientId, messageId, guid);
        m_inflightStore.put(clientId, m);
    }

    @Override
    public void bindToDeliver(MessageGUID guid, String clientId) {
        List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientId, new ArrayList<MessageGUID>());
        guids.add(guid);
        m_enqueuedStore.put(clientId, guids);
        messagesStore.incUsageCounter(guid);
    }

    @Override
    public Collection<MessageGUID> enqueued(String clientId) {
        return Utils.defaultGet(m_enqueuedStore, clientId, new ArrayList<MessageGUID>());
    }

    @Override
    public void removeMessageGUID(String clientId, int messageId, MessageGUID guid) {
        if (guid != null) {

            //if counter gets to 0 then remove from storage
            messagesStore.removeStoredMessage(guid);
            SessionsStoreUtils.removeMessageToGuid(clientId, messageId);
        }
    }

    @Override
    public void removeEnqueued(String clientId, int messageId, MessageGUID guid) {

        List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientId, new ArrayList<MessageGUID>());
        guids.remove(guid);
        m_enqueuedStore.put(clientId, guids);
        messagesStore.decUsageCounter(guid);
        //if counter gets to 0 then remove from storage
        messagesStore.removeStoredMessage(guid);
        SessionsStoreUtils.removeMessageToGuid(clientId, messageId);
    }

    @Override
    public void moveInFlightToSecondPhaseAckWaiting(String clientId, int messageId) {
        LOG.info("acknowledging inflight clientId <{}> messageID {}", clientId, messageId);
        Map<Integer, MessageGUID> m = m_inflightStore.get(clientId);
        if (m == null) {
            LOG.error("Can't find the inFlight record for client <{}>", clientId);
            return;
        }
        MessageGUID guid = m.remove(messageId);

        //remove from the ids store
        Set<Integer> inFlightForClient = m_inFlightIds.get(clientId);
        if (inFlightForClient != null) {
            inFlightForClient.remove(messageId);
        }

        LOG.info("Moving to second phase store");
        Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientId, new HashMap<Integer, MessageGUID>());
        messageIDs.put(messageId, guid);
        m_secondPhaseStore.put(clientId, messageIDs);
    }

    @Override
    public MessageGUID secondPhaseAcknowledged(String clientId, int messageId) {
        Map<Integer, MessageGUID> messageIDs = Utils.defaultGet(m_secondPhaseStore, clientId, new HashMap<Integer, MessageGUID>());
        MessageGUID guid = messageIDs.remove(messageId);
        m_secondPhaseStore.put(clientId, messageIDs);
        return guid;
    }

    @Override
    public MessageGUID mapToGuid(String clientId, int messageId) {
        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = redissonClient.getMap(SessionsStoreUtils.messageId2GuidsMapName(clientId));
        return messageIdToGuid.get(messageId);
    }

    @Override
    public StoredMessage getInflightMessage(String clientId, int messageId) {
        Map<Integer, MessageGUID> clientEntries = m_inflightStore.get(clientId);
        if (clientEntries == null) {
            return null;
        }
        MessageGUID guid = clientEntries.get(messageId);
        LOG.info("inflight messageID {} guid <{}>", messageId, guid);
        if (guid == null) {
            return null;
        }
        return messagesStore.getMessageByGuid(guid);
    }


}
